package cn.getech.data.development.controller.real;

import cn.getech.data.development.constant.TableDataHeaderType;
import cn.getech.data.development.constant.YesOrNoEnum;
import cn.getech.data.development.entity.flink.FlinkNodeField;
import cn.getech.data.development.model.vo.FlinkTaskNodeVO;
import cn.getech.data.development.service.FlinkTaskNodeService;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.R;
import cn.getech.system.center.annotation.SysLog;
import cn.getech.system.center.constant.OperationModulesEnum;
import cn.getech.system.center.constant.OperationTypeEnum;
import cn.getech.system.center.constant.module.ModuleEnum;
import cn.hutool.core.text.csv.CsvData;
import cn.hutool.core.text.csv.CsvReader;
import cn.hutool.core.text.csv.CsvRow;
import cn.hutool.core.text.csv.CsvUtil;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import javax.servlet.http.HttpServletResponse;
import java.io.File;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

@Slf4j
@Api(tags = "实时任务")
@RestController
@RequestMapping("/bdp/realTime/task/node")
public class RealTimeFlinkController {

    @Autowired
    private FlinkTaskNodeService taskNodeService;

    @ApiOperation("保存节点")
    @PostMapping("/save")
    @SysLog(value="保存实时任务节点",type= OperationTypeEnum.ADDCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.TASK_OPERATION_MODULE_CODE)
    public R saveNode(@RequestBody FlinkTaskNodeVO nodeVO) {
        String msg = nodeVO.verificationParam();
        if (StringUtils.isNotEmpty(msg)) {
            return R.error(msg);
        }
        taskNodeService.saveNode(nodeVO);
        return R.okWithData(nodeVO.getId());
    }

    @ApiOperation("保存节点")
    @PostMapping("/delete")
    @SysLog(value="删除实时任务节点",type= OperationTypeEnum.DELETECODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.TASK_OPERATION_MODULE_CODE)

    public R removeNode(@RequestBody FlinkTaskNodeVO nodeVO) {
        taskNodeService.removeNode(nodeVO);
        return R.ok();
    }
    @ApiOperation("节点详情")
    @GetMapping("/info")
    public R getDetailById(Integer id, @RequestParam(required = false)  Integer preFlag) {
        if (id == null) {
            return R.error("节点不存在");
        }
        FlinkTaskNodeVO vo = taskNodeService.getDetailById(id);
        if (YesOrNoEnum.YES.getValue().equals(preFlag)) {
            for (FlinkNodeField flinkNodeField : vo.getFieldList()) {
                if (StringUtils.isNotEmpty(flinkNodeField.getFieldAlias())) {
                    flinkNodeField.setFieldName(flinkNodeField.getFieldAlias());
                }
            }
        }
        vo.setSql(vo.wrapSQL());
        return R.okWithData(vo);
    }

    @ApiOperation("节点详情")
    @GetMapping("/field/type")// /data-development/bdp/realTime/task/node/field/type
    public R fieldType()  {
       List<String> vo = taskNodeService.fieldType();
        return R.okWithData(vo);
    }

    @ApiOperation("sql预览")
    @PostMapping("/wrap/sql")   //  /data-development/bdp/realTime/task/node/wrap/sql
    public R wrapSQL(@RequestBody FlinkTaskNodeVO nodeVO) {
        String msg = nodeVO.verificationParam();
        if (StringUtils.isNotEmpty(msg)) {
            return R.error(msg);
        }
        taskNodeService.saveNode(nodeVO);
        String sql = nodeVO.wrapSQL();
        return R.okWithData(sql);
    }


    @ApiOperation("下载模版")
    @SysLog(value = "下载模板", type = OperationTypeEnum.DOWNLOADCODE, modulesType = OperationModulesEnum.TABLEMODELCODE
            ,platformModuleType = ModuleEnum.TASK_OPERATION_MODULE_CODE)
    @GetMapping("/template") //  /data-development/bdp/realTime/task/node/template
    public void downloadField(HttpServletResponse response) throws Exception {
        response.reset();
        String fileName = String.format("field_data.csv");
        response.setHeader("Content-Disposition", "attachment;filename=" + fileName);
        response.setHeader("filename", fileName);
        OutputStream outputStream = response.getOutputStream();
        String header = String.format("%s,%s,%s",
                TableDataHeaderType.fieldName.getDesc(),
                TableDataHeaderType.fieldType.getDesc(),
                TableDataHeaderType.fieldAlias.getDesc()
                );
        outputStream.write(header.getBytes("GBK"));
        outputStream.close();
    }

    @ApiOperation("上传字段")
    @SysLog(value = "上传字段", type = OperationTypeEnum.UPLOADCODE, modulesType = OperationModulesEnum.TABLEMODELCODE
            ,platformModuleType = ModuleEnum.TASK_OPERATION_MODULE_CODE)
    @PostMapping("/uploadField")//  /data-development/bdp/realTime/task/node/uploadField
    public R uploadField(@RequestParam(value = "file", required = false) MultipartFile multipartFile) {
        if (multipartFile == null) {
            return R.error("您还没有选择要上传的文件");
        }
        if (!multipartFile.isEmpty()) {
            List<FlinkNodeField> fieldList=new ArrayList<>();
            try {
                File file = File.createTempFile("tmp", null);
                String fileName=multipartFile.getOriginalFilename();
                multipartFile.transferTo(file);
                file.deleteOnExit();
                if (fileName.contains(".csv")){
                    readFiledForCsv(file,fieldList);
                }
            } catch (RRException e) {
                return R.error(e.getMsg());
            } catch (Exception e) {
                log.error("文件解析错误：",e);
                return R.error("请按照模板格式修改上传文件");
            }
            return R.okWithData(fieldList);
        } else {
            return R.error("文件为空");
        }
    }
    private  void readFiledForCsv(File file , List<FlinkNodeField> nodeFields) throws RRException{
        // file.getBytes()
        CsvReader reader = CsvUtil.getReader();
        // 从文件中读取CSV数据
        CsvData data = reader.read(file, Charset.forName("GBK"));
        List<String> filedTtypes = taskNodeService.fieldType();
        List<CsvRow> rows = data.getRows();
        if (rows != null && rows.size() >= 2) {
            //遍历行
            for (int i = 1; i < rows.size(); i++) {
                for (int j = 0; j < 3; j++) {
                    if (null == rows.get(i).get(j) || "".equals(rows.get(i).get(j))) {
                        throw new RRException(String.format("表格中  '%s'  存在空数据", TableDataHeaderType.ObjOf(j).getDesc()));
                    }
                }
                FlinkNodeField nodeField = new FlinkNodeField();
                nodeField.setFieldName(rows.get(i).get(TableDataHeaderType.fieldName.getCode()));
                String type = rows.get(i).get(TableDataHeaderType.fieldType.getCode());
                if (!filedTtypes.contains(type.toLowerCase())) {
                    throw new RRException(String.format("字段类型%s的字段不合规", rows.get(i).get(TableDataHeaderType.fieldType.getCode())));
                }
                nodeField.setFieldType(rows.get(i).get(TableDataHeaderType.fieldType.getCode()));
                nodeField.setFieldDesc(rows.get(i).get(TableDataHeaderType.fieldAlias.getCode()));
                nodeField.setFieldOrder(i + 1);
                nodeFields.add(nodeField);
            }
        }

    }
}
